#include "EventLoopMgr.h"
#include <sstream>
#ifdef WIN32
#include <windows.h>
#else
#include <pthread.h>
#endif
EventLoopMngr* EventLoopMngr::m_oInstance = NULL;
std::once_flag EventLoopMngr::m_flag;
class Event
{
public:
	int32_t Wait()
	{
		std::unique_lock<decltype(m_MutexCondition)> lock{ m_MutexCondition };
		m_Condition.wait(lock, [&]()->bool { return ret != -1; }); // suspend and wait ...	
		return ret;
	}
	void Set()
	{
		std::unique_lock<decltype(m_MutexCondition)> lock(m_MutexCondition);
		ret = 0;
		m_Condition.notify_one();
	}
private:
	std::mutex m_MutexCondition;
	std::condition_variable m_Condition;
	int32_t ret = -1;
};
TaskScheduler::TaskScheduler()
{
}
TaskScheduler::~TaskScheduler()
{
	StopRun();
}

int64_t TaskScheduler::getCurrentTicket()
{
	//std::chrono::steady_clock::time_point t1 = std::chrono::steady_clock::now();
	//std::chrono::steady_clock::time_point t11;
	//int64_t duration = std::chrono::duration_cast<std::chrono::microseconds>(t1 - t11).count();
	//int64_t duration = std::chrono::duration_cast<std::chrono::milliseconds>(t1 - t11).count();
	return  std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now().time_since_epoch()).count();
}
int32_t TaskScheduler::StartRun(std::string threadName)
{
	m_loopThread = new std::thread(&TaskScheduler::RunLoop, this);
	//std::unique_lock<std::mutex> lock(_mutex);
	//_condition.wait(lock, [this]() { return m_bStart == true; });
	while (!m_bStart)
	{
		std::this_thread::sleep_for(std::chrono::milliseconds(1));
	}
	SetThreadName(threadName);
	if (m_loopThread)
	{
		return 0;
	}
	return -1;

}
void TaskScheduler::StopRun()
{
	m_bExitThread = true;
	{
		std::unique_lock<std::mutex> lock(_mutex);
		_queue.push_back([]() {int a; });
	}
	_condition.notify_one();

	if (m_loopThread)
	{
		m_loopThread->join();
		delete m_loopThread;
		m_loopThread = 0;
	}
}

int64_t TaskScheduler::AddTimer(int64_t iDelay, std::function<bool()> fcb)
{

	//printf("### EventLoop TaskScheduler::AddTimer Start iDelay:%lld ThreadID:%s this:%p \n", iDelay, GetThreadId().c_str(), this);
	//MEDIA_TRACE_LOG("EventLoop", LL_INFO, "### TaskScheduler::AddTimer Start iDelay:%lld ThreadID:%s this:%p ", iDelay, GetThreadId().c_str(), this);
	Event tEvent;
	auto t_timer = std::make_shared<Timer>(iDelay, fcb);
	int64_t t_timer_id = 0;
	InvokTask([this, &t_timer, &t_timer_id]()
	{
		//printf("*** EventLoop TaskScheduler::AddTimer InvokTask Start   ThreadID:%s this:%p \n",  GetThreadId().c_str(), this);
		//MEDIA_TRACE_LOG("EventLoop", LL_INFO, "*** TaskScheduler::AddTimer InvokTask Start   ThreadID:%s this:%p ", GetThreadId().c_str(), this);

		m_iTimerId++;
		t_timer->m_iTimerId = m_iTimerId;
		t_timer_id = m_iTimerId;
		uint64_t time_line = getCurrentTicket() + t_timer->m_iDelay;
		m_delayTask.emplace(time_line, std::move(t_timer));

		//printf("*** EventLoop TaskScheduler::AddTimer InvokTask End ThreadID:%s this:%p \n",  GetThreadId().c_str(), this);
		//MEDIA_TRACE_LOG("EventLoop", LL_INFO, "*** TaskScheduler::AddTimer InvokTask End ThreadID:%s this:%p ", GetThreadId().c_str(), this);
	}
	);

	//printf("### EventLoop TaskScheduler::AddTimer End iDelay:%lld ThreadID:%s t_timer_id:%lld this:%p \n", iDelay, GetThreadId().c_str(), t_timer_id, this);
	//MEDIA_TRACE_LOG("EventLoop", LL_INFO, "### TaskScheduler::AddTimer End iDelay:%lld ThreadID:%s t_timer_id:%lld this:%p ", iDelay, GetThreadId().c_str(), t_timer_id, this);
	return t_timer_id;
}

void TaskScheduler::PostTask(Task task)
{
	// 调用线程和将要打入的线程是同一个
	if (_loopThreadId == std::this_thread::get_id())
	{
		// 原地执行就可以
		task();
		return;
	}
	{
		std::unique_lock<std::mutex> lock(_mutex);
		_queue.push_back([task]() {
			task();

		});
		_condition.notify_one();
	}
	
}

void TaskScheduler::InvokTask(Task task)
{
	// 调用线程和将要打入的线程是同一个
	if (_loopThreadId == std::this_thread::get_id())
	{
		// 原地执行就可以
		task();
		return;
	}

	Event tEvent;
	{
		std::unique_lock<std::mutex> lock(_mutex);
		_queue.push_back([task, &tEvent]() {
			task();
			tEvent.Set();
		});
		_condition.notify_one();
	}
	
	tEvent.Wait();
}

std::string TaskScheduler::GetThreadId()
{
	std::ostringstream oss;
	oss << _loopThreadId;
	return oss.str();
}

int32_t TaskScheduler::DelTimer(int64_t timerId)
{
	//printf("### EventLoop TaskScheduler::DelTimer Start timerId:%lld ThreadID:%s this:%p \n", timerId, GetThreadId().c_str(), this);
	//MEDIA_TRACE_LOG("EventLoop", LL_INFO, "### TaskScheduler::DelTimer Start timerId:%lld ThreadID:%s this:%p ", timerId, GetThreadId().c_str(), this);
	int ret = -1;
	InvokTask([this, timerId, &ret]()
	{
		//printf("*** EventLoop TaskScheduler::DelTimer Start InvokTask timerId:%lld ThreadID:%s this:%p \n", timerId, GetThreadId().c_str(), this);
		//MEDIA_TRACE_LOG("EventLoop", LL_INFO, "*** TaskScheduler::DelTimer Start InvokTask timerId:%lld ThreadID:%s this:%p ", timerId, GetThreadId().c_str(), this);
		std::multimap<uint64_t, std::shared_ptr<Timer> >::iterator it = m_delayTask.begin();
		for (; it != m_delayTask.end();)
		{
			std::shared_ptr<Timer> pTimer = it->second;
			if (pTimer->m_iTimerId == timerId)
			{
				it = m_delayTask.erase(it);
				ret = 0;
				break;
			}
			else
			{
				it++;
			}
		}

		//printf("*** EventLoop TaskScheduler::DelTimer End InvokTask timerId:%lld ThreadID:%s ret:%d this:%p \n", timerId, GetThreadId().c_str(), ret, this);
		//MEDIA_TRACE_LOG("EventLoop", LL_INFO, "*** TaskScheduler::DelTimer End InvokTask timerId:%lld ThreadID:%s ret:%d this:%p ", timerId, GetThreadId().c_str(), ret, this);

	});

	//printf("### EventLoop TaskScheduler::DelTimer End timerId:%lld ThreadID:%s ret:%d this:%p \n", timerId, GetThreadId().c_str(), ret, this);
	//MEDIA_TRACE_LOG("EventLoop", LL_INFO, "### TaskScheduler::DelTimer End timerId:%lld ThreadID:%s ret:%d this:%p ", timerId, GetThreadId().c_str(), ret, this);
	return ret;
}


void TaskScheduler::RunLoop()
{
	_loopThreadId = std::this_thread::get_id();
	m_bStart = true;
	//_condition.notify_one();

	while (!m_bExitThread)
	{
		int64_t delay = getMinDelay(); // 处理定时器任务

		std::list<Task> t_queue;
		{
			std::unique_lock<std::mutex> lock(_mutex);
			////printf("delay:%lld this:%p \n", delay, this);
			//_condition.wait_for(lock, std::chrono::milliseconds(delay), [this]() { return !_queue.empty(); });
			_condition.wait_for(lock, std::chrono::milliseconds(delay), [this]() { return !_queue.empty(); });

			////printf("_queue.size:%d \n", _queue.size());
			if (!_queue.empty())
			{
				t_queue.swap(_queue);

			}
		}
		// 处理任务
		if (!t_queue.empty())
		{
			for (std::list<Task>::iterator it = t_queue.begin(); it != t_queue.end(); it++)
			{
				Task task = *it;
				task();
			}
			t_queue.clear();
		}
	}

}
void TaskScheduler::SetThreadName(std::string threadName)
{
#ifdef WIN32

	int wcharCount = MultiByteToWideChar(CP_UTF8, 0, threadName.c_str(), -1, NULL, 0);
	std::wstring wideString(wcharCount, L'\0');
	MultiByteToWideChar(CP_UTF8, 0, threadName.c_str(), -1, &wideString[0], wcharCount);

	PCWSTR pwstr = wideString.c_str();
	HRESULT hr = SetThreadDescription(m_loopThread->native_handle(), pwstr);
	if (FAILED(hr))
	{
		// Call failed.
		//MEDIA_TRACE_LOG("EventLoop", LL_ERROR, "*** TaskScheduler::SetThreadName error this:%p ",this);
	}
#else
	pthread_setname_np(m_loopThread->native_handle(), threadName.c_str());
#endif
}

int64_t TaskScheduler::getMinDelay() {
	auto it = m_delayTask.begin();
	if (it == m_delayTask.end()) {
		return 10;
	}
	auto now = getCurrentTicket();
	if (it->first > now) {
		//所有任务尚未到期
		////printf("it->first:%llu now:%lld dif:%lld \n", it->first, now, it->first - now);
		return it->first - now;
	}
	//执行已到期的任务并刷新休眠延时
	return flushDelayTask(now);
}

int64_t TaskScheduler::flushDelayTask(uint64_t now_time) {

	////printf("############# sleep :%llu \n", now_time);
	decltype(m_delayTask) taskCopy;
	taskCopy.swap(m_delayTask);
	uint64_t start = getCurrentTicket();
	for (auto it = taskCopy.begin(); it != taskCopy.end() && it->first <= now_time; it = taskCopy.erase(it)) {
		//已到期的任务
		//auto next_delay = (*(it->second))();
		std::shared_ptr<Timer>  pTimer = it->second;
		if (pTimer->m_cb)
		{
			bool ret = pTimer->m_cb();
			if (ret)
			{
				m_delayTask.emplace(pTimer->m_iDelay + now_time, std::move(it->second));
			}
			else
			{
				//MEDIA_TRACE_LOG("EventLoop", LL_INFO, "TaskScheduler::flushDelayTask this:%p m_iTimerId:%lld ret:%d", this, pTimer->m_iTimerId, ret);
			}
		}
	}
	uint64_t end = getCurrentTicket();
	int64_t useTime = end - start;
	if (useTime > 11)
	{
		////printf("timers_poll::flushDelayTask useTimer = %d \n", useTime);
	}

	taskCopy.insert(m_delayTask.begin(), m_delayTask.end());
	taskCopy.swap(m_delayTask);

	auto it = m_delayTask.begin();
	if (it == m_delayTask.end()) {
		//没有剩余的定时器了
		return 0;
	}
	//最近一个定时器的执行延时

	int64_t t_sleep = it->first - getCurrentTicket();
	if (t_sleep < -1)
	{
		//printf("timers_poll::flushDelayTask  t_sleep = %lld  \n", t_sleep);
	}
	return t_sleep > 0 ? t_sleep : 0;

}


EventLoop::EventLoop()
{
	num_threads_ = 1;

}

EventLoop::~EventLoop()
{
	this->Quit();
}

std::shared_ptr<TaskScheduler> EventLoop::GetTaskScheduler()
{
	std::lock_guard<std::mutex> locker(mutex_);

	if (task_schedulers_.size() == 0)
	{
		return NULL;
	}

	if (task_schedulers_.size() == 1) {
		return task_schedulers_.at(0);
	}
	else {
		auto task_scheduler = task_schedulers_.at(index_);
		index_++;
		if (index_ >= task_schedulers_.size()) {
			index_ = 0;
		}
		return task_scheduler;
	}

	return nullptr;
}


int32_t EventLoop::Loop(std::string name_thread, uint32_t num_threads)
{
	std::lock_guard<std::mutex> locker(mutex_);
	m_name = name_thread;
	num_threads_ = num_threads;
	//MEDIA_TRACE_LOG("EventLoop", LL_INFO, "### EventLoop::Loop Start this:%p num_threads_:%u name:%s ", this, num_threads_, m_name.c_str());
	if (!task_schedulers_.empty()) {
		//printf("### EventLoop Have Looped !task_schedulers_.empty() this:%p task_schedulers_.size():%d ", this, task_schedulers_.size());
		//MEDIA_TRACE_LOG("EventLoop", LL_INFO, "### EventLoop End Have Looped !task_schedulers_.empty() this:%p task_schedulers_.size():%d ", this, task_schedulers_.size());
		return task_schedulers_.size();
	}

	for (int32_t n = 0; n < num_threads_; n++)
	{
		//printf("*** EventLoop::Loop StartRun before n:%d this:%p num_threads_:%u \n", n, this, num_threads_);
		//MEDIA_TRACE_LOG("EventLoop", LL_INFO, "*** EventLoop::Loop StartRun before n:%d this:%p ", n, this);
		std::string t_thread_name = m_name + "_" + std::to_string(n);
		int32_t ret = 0;
		std::shared_ptr<TaskScheduler> task_scheduler_ptr(new TaskScheduler());
		ret = task_scheduler_ptr->StartRun(t_thread_name);
		if (0 == ret)
		{
			task_schedulers_.push_back(task_scheduler_ptr);
		}
		//printf("*** EventLoop::Loop StartRun end n:%d ret:%d task_scheduler_ptr:%p ThreadID:%s this:%p \n", n, ret, task_scheduler_ptr.get(), task_scheduler_ptr->GetThreadId().c_str(), this);
		//MEDIA_TRACE_LOG("EventLoop", LL_INFO, "*** EventLoop::Loop StartRun end n:%d ret:%d task_scheduler_ptr:%p ThreadID:%s this:%p ", n, ret, task_scheduler_ptr.get(), task_scheduler_ptr->GetThreadId().c_str(), this);
	}

	//MEDIA_TRACE_LOG("EventLoop", LL_INFO, "### EventLoop::Loop End this:%p task_schedulers_.size():%d ", this, task_schedulers_.size());
	return task_schedulers_.size();

}

void EventLoop::Quit()
{
	//printf("EventLoop::Quit Start this:%p task_schedulers_.size():%d \n", this, task_schedulers_.size());
	////MEDIA_TRACE_LOG("EventLoop", LL_INFO, "### EventLoop::Quit Start this:%p task_schedulers_.size():%d ", this, task_schedulers_.size());
	int i = 0;
	for (auto iter : task_schedulers_) {
		std::shared_ptr<TaskScheduler> t_scheduler = iter;
		//printf("EventLoop::StopRun before i:%d iter:%p ThreadID:%s this:%p \n", i, t_scheduler.get(), t_scheduler->GetThreadId().c_str(), this);
		////MEDIA_TRACE_LOG("EventLoop", LL_INFO, "*** EventLoop::StopRun before i:%d iter:%p ThreadID:%s this:%p ", i, t_scheduler.get(), t_scheduler->GetThreadId().c_str(), this);
		iter->StopRun();
		//printf("EventLoop::StopRun after i:%d iter:%p ThreadID:%s this:%p \n", i, t_scheduler.get(), t_scheduler->GetThreadId().c_str(), this);
		////MEDIA_TRACE_LOG("EventLoop", LL_INFO, "*** EventLoop::StopRun after i:%d iter:%p ThreadID:%s this:%p ", i, t_scheduler.get(), t_scheduler->GetThreadId().c_str(), this);
		i++;
	}
	task_schedulers_.clear();

	//printf("EventLoop::Quit End this:%p \n", this);
	////MEDIA_TRACE_LOG("EventLoop", LL_INFO, "### EventLoop::Quit End this:%p ", this);

}
EventLoopMngr* EventLoopMngr::GetInstance()
{
	//return &(EventLoopMngr::m_oInstance);
	std::call_once(m_flag, [&](){
		EventLoopMngr::m_oInstance = new EventLoopMngr();
		printf("\n *********** ########## EventLoopMngr::GetInstance call_once \n");
	});
	if (EventLoopMngr::m_oInstance)
	{
		return EventLoopMngr::m_oInstance;
	}
	return NULL;
}

EventLoopMngr::EventLoopMngr()
{

}
EventLoopMngr::~EventLoopMngr()
{
	m_oEventLoop.Quit();
}
int32_t EventLoopMngr::Init(uint32_t num_threads, std::string name_thread)
{
	int32_t ret = m_oEventLoop.Loop(name_thread, num_threads);
	return ret;
}

int32_t EventLoopMngr::InitRender(uint32_t num_threads, std::string name_thread)
{
	int32_t ret = m_oEventLoopRender.Loop(name_thread,num_threads);
	return ret;
}

